use std::any::Any;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use async_trait::async_trait;
use tokio::sync::mpsc;
use tokio::time::sleep;

use protoactor_rs::actor::{Actor, ActorContext, ActorError, DefaultActorContext, Props};
// 导入将要实现的远程模块
// use protoactor_rs::remote::RemoteActorSystem;

/// 消息数量
const MESSAGE_COUNT: usize = 1_000_000;

/// 批处理大小
const BATCH_SIZE: usize = 100;

/// 远程基准测试消息
#[derive(Debug, Clone)]
struct BenchmarkMessage {
    message_id: usize,
}

/// 远程基准测试响应
#[derive(Debug, Clone)]
struct BenchmarkResponse {
    message_id: usize,
}

/// 远程基准测试启动消息
#[derive(Debug, Clone)]
struct StartBenchmark {
    count: usize,
}

/// 远程基准测试完成消息
#[derive(Debug, Clone)]
struct BenchmarkComplete {
    elapsed: Duration,
    messages_per_second: f64,
}

/// 本地基准客户端Actor，用于发送消息和计算性能
struct ClientActor {
    counter: AtomicUsize,
    report_tx: mpsc::Sender<BenchmarkComplete>,
    start_time: Option<Instant>,
}

impl Clone for ClientActor {
    fn clone(&self) -> Self {
        Self {
            counter: AtomicUsize::new(self.counter.load(Ordering::Relaxed)),
            report_tx: self.report_tx.clone(),
            start_time: self.start_time.clone(),
        }
    }
}

#[async_trait]
impl Actor for ClientActor {
    type Context = DefaultActorContext;

    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn Any + Send>) -> Result<(), ActorError> {
        if let Some(_) = msg.downcast_ref::<StartBenchmark>() {
            println!("客户端开始基准测试，将发送 {} 条消息", MESSAGE_COUNT);
            
            // 记录开始时间
            let start_time = Instant::now();
            
            // 更新状态
            let mut this = self.clone();
            this.start_time = Some(start_time);
            
            // TODO: 获取远程引用
            // ...
            
            println!("向远程服务器发送消息...");
            
            // 发送消息（注意：这仅为占位，实际应使用远程PID）
            // 批量处理，减少输出量
            for batch in 0..(MESSAGE_COUNT / BATCH_SIZE) {
                let start = batch * BATCH_SIZE;
                let end = (batch + 1) * BATCH_SIZE;
                
                for i in start..end {
                    let msg = BenchmarkMessage { message_id: i };
                    
                    // 这里应该向远程PID发送消息
                    // remote_system.send(&remote_pid, msg).await?;
                    
                    // 仅为测试占位
                    if i % 100_000 == 0 {
                        println!("已发送 {} 条消息", i);
                    }
                }
                
                // 添加短暂延迟，避免过快消耗内存
                if batch % 10 == 0 {
                    sleep(Duration::from_millis(1)).await;
                }
            }
        }
        else if let Some(response) = msg.downcast_ref::<BenchmarkResponse>() {
            // 收到一个响应
            let count = self.counter.fetch_add(1, Ordering::Relaxed) + 1;
            
            // 每收到10万条消息输出一次
            if count % 100_000 == 0 {
                println!("收到 {} 条响应", count);
            }
            
            // 检查是否完成所有消息
            if count >= MESSAGE_COUNT {
                if let Some(start_time) = self.start_time {
                    let elapsed = start_time.elapsed();
                    let msgs_per_sec = MESSAGE_COUNT as f64 / elapsed.as_secs_f64();
                    
                    println!("基准测试完成!");
                    println!("总消息数: {}", MESSAGE_COUNT);
                    println!("耗时: {:?}", elapsed);
                    println!("每秒消息量: {:.2} msgs/sec", msgs_per_sec);
                    
                    // 报告结果
                    let result = BenchmarkComplete {
                        elapsed,
                        messages_per_second: msgs_per_sec,
                    };
                    
                    if let Err(e) = self.report_tx.send(result).await {
                        println!("发送结果失败: {}", e);
                    }
                }
            }
        }
        
        Ok(())
    }
}

/// 远程基准服务器Actor，接收消息并回复
struct ServerActor;

#[async_trait]
impl Actor for ServerActor {
    type Context = DefaultActorContext;

    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn Any + Send>) -> Result<(), ActorError> {
        if let Some(benchmark) = msg.downcast_ref::<BenchmarkMessage>() {
            // 创建响应
            let response = BenchmarkResponse {
                message_id: benchmark.message_id,
            };
            
            // 回复发送者
            ctx.respond(response);
            
            // 每收到100万条消息打印一次日志
            if benchmark.message_id % 1_000_000 == 0 {
                println!("服务器处理了 {} 条消息", benchmark.message_id);
            }
        }
        
        Ok(())
    }
    
    async fn started(&self, _ctx: &mut Self::Context) -> Result<(), ActorError> {
        println!("基准测试服务器已启动");
        Ok(())
    }
}

/// 服务器端入口
async fn run_server() -> Result<(), Box<dyn std::error::Error>> {
    println!("启动远程基准测试服务器...");
    
    // 创建系统
    let system = protoactor_rs::system::ActorSystem::new()?;
    
    // 创建服务器Actor
    let props = Props::from_producer(|| ServerActor);
    let server_pid = system.spawn::<ServerActor>(props).await?;
    
    println!("服务器准备就绪，等待连接和消息");
    
    // 持续运行，直到中断
    loop {
        sleep(Duration::from_secs(1)).await;
    }
}

/// 客户端入口
async fn run_client() -> Result<(), Box<dyn std::error::Error>> {
    println!("启动远程基准测试客户端...");
    
    // 创建系统
    let system = protoactor_rs::system::ActorSystem::new()?;
    
    // 创建结果通道
    let (tx, mut rx) = mpsc::channel(10);
    
    // 创建客户端Actor
    let client = ClientActor {
        counter: AtomicUsize::new(0),
        report_tx: tx,
        start_time: None,
    };
    
    let props = Props::from_producer(move || client.clone());
    let client_pid = system.spawn::<ClientActor>(props).await?;
    
    println!("客户端准备就绪");
    
    // 发送启动消息
    let start_msg = StartBenchmark {
        count: MESSAGE_COUNT,
    };
    
    system.send(&client_pid, start_msg).await?;
    
    // 等待测试结果
    if let Some(result) = rx.recv().await {
        println!("\n基准测试结果:");
        println!("耗时: {:?}", result.elapsed);
        println!("每秒消息量: {:.2} msgs/sec", result.messages_per_second);
    } else {
        println!("未收到测试结果");
    }
    
    // 停止客户端
    system.stop(&client_pid).await?;
    
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("远程基准测试");
    println!("注意：目前远程通信功能尚未实现，这只是一个框架示例");
    
    // 解析命令行参数
    let args: Vec<String> = std::env::args().collect();
    
    if args.len() < 2 {
        println!("用法: {} [client|server]", args[0]);
        return Ok(());
    }
    
    match args[1].as_str() {
        "server" => run_server().await,
        "client" => run_client().await,
        _ => {
            println!("未知的角色: {}", args[1]);
            println!("用法: {} [client|server]", args[0]);
            Ok(())
        }
    }
} 